package com.demo.producer;

import com.alibaba.fastjson.JSON;
import com.demo.common.Constant;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

import java.nio.charset.StandardCharsets;

/**
 * RocketMQ发送消息方式:
 * 业务：1.处理本服务业务逻辑 2.发送消息  3.处理减少操作
 * <p>
 * 1.同步发送:消息发送到mq中必须要等mq持久化成功以后才能够执行后面的业务逻辑
 * 2.异步发送:消息发送到mq中不需要等mq持久化成功以后才能够执行后面的业务逻辑(明显异步的性能高些)
 * 3.一次性发送:消息发送到mq中,就可以执行后面业务逻辑了,注意一次性发送没有返回结果(不关心信息有没有保留成功)
 *
 * @author yexi
 */
public class SyncProducer {
    public static void main(String[] args) throws Exception {
        //1、实例化一个信息生产者对象，并且指定一个生产者组(一个组里有多个生产者，多个生产者可在同一个组里面)
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer(Constant.PRODUCER_GROUP_1);
        //2、设置名字服务的地址(与 NameServer 进行绑定)
        //如果出现 No route info of this topic 错误时，就是生产者找不到 rocketmq 了，可能是端口号写错了亦或者 mq 服务挂掉了(重启下)
        defaultMQProducer.setNamesrvAddr(Constant.ROCKET_MQ_ADDRESS);
        //3、启动生产者
        defaultMQProducer.start();

        // topic。主题。
        String topic = "01-sync";
        // 标签，可选参数。生产者只支持贴一个标签，但是消费者可以获取多个标签。
        String tags = "tagA";
        // 要发送的消息数据。
        String msg = "RocketMQ Sync Send ";

        //4、创建消息对象，把数据封装到消息对象中。可以指定主题、标签和信息体。
        for (int i = 0; i < 100; i++) {
            Message message = new Message(
                    topic,
                    tags,
                    (msg + i).getBytes(StandardCharsets.UTF_8)
            );
            //5、发送消息。send(msg)为"同步发送消息"。
            SendResult sendResult = defaultMQProducer.send(message);
            System.out.println("发送后返回的消息: " + JSON.toJSONString(sendResult));
            //没输出表示失败了
            System.out.println("成功发送的消息ID: " + sendResult.getMsgId());
            System.out.println("----------------------------");
        }

        //6、关闭连接(关闭资源)，不再发送消息。
        defaultMQProducer.shutdown();
    }
}
